package org.example

import org.apache.spark.sql.SparkSession

import java.util.regex.Pattern

object SparkYun1_movies {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("sparkBase")
      .getOrCreate()
    val sc = spark.sparkContext
//打印电影数据概况
    val filePath = "src/main/resources/"
    val usersRDD = sc.textFile(filePath + "users.dat")
    val moviesRDD = sc.textFile(filePath + "movies.dat")
    val ratingsRDD = sc.textFile(filePath + "ratings.dat")
    val occupationsRDD = sc.textFile(filePath + "occupations.dat")
    println("用户总数：" + usersRDD.count())
    println("电影总数：" + moviesRDD.count())
    println("评分数：" + ratingsRDD.count())
    println("职业数：" + occupationsRDD.count())

//    分析一：统计某用户评分的电影总数和对应的电影详情
    val userMovieRating = ratingsRDD.map(_.split("::"))
      .map(tp => (tp(1),tp(0)))  //转换成key(电影Id) value(用户Id)
      .filter(_._2.equals("8"))// 过滤出某Id的电影评分总数
//    println("用户Id为8的电影评分总数是:" + userMovieRating.count() )
//    将电影详情转换为key value型
    val movieInfo = moviesRDD.map(_.split("::"))
      .map(sp => (sp(0),(sp(1),sp(2))))
//    连接某用户评分表和电影详情表，并重新映射成新的表
    val detail = userMovieRating.join(movieInfo).map(xp => {
      (xp._1,xp._2._1,xp._2._2._1,xp._2._2._1)
    })
//      detail.take(5).foreach(println)
//    分析二：找出评分最高的10部电影和观影人次最高的10部电影

//    分析三:统计电影类型(有多少部恐怖片)
      val movieStyle = moviesRDD.map(_.split("::"))
        .map(x => (x(0),x(2))) //取电影Id 电影类型
        .flatMapValues(x => x.split("\\|")) //根据间隔符号拆分
        .map(x => (x._2, 1))
        .reduceByKey((x,y) => x + y)
    //练习:在控制台输出:喜剧片的数量为1200部
    movieStyle.foreach(println)
//    分析四:统计各年度电影数量
      val pattern = Pattern.compile("(.*)(\\(\\d{4}\\))")
    val movieYear = moviesRDD.map(_.split("::"))
      .map(x => (x(1), 1))
      .map(x => {
        var name = ""
        var year = ""
        val matcher = pattern.matcher(x._1)
        if (matcher.find()) {
          name = matcher.group(1)
          year = matcher.group(2)
          year = year.substring(1,year.length() - 1)
        }
        if (year == "") {
          (-1, 1)
        } else {
          (year.toInt, 1)
        }
      })
      .reduceByKey((x,y) => x + y)
      .sortByKey()
      .collect()
    movieYear.foreach(println)
//    在控制台打印2000年度的电影总数
    sc.stop()
  }
}
